package top.doe.flink.demos;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import lombok.AllArgsConstructor;
import lombok.Data;
import lombok.NoArgsConstructor;
import org.apache.flink.api.common.functions.CoGroupFunction;
import org.apache.flink.api.common.functions.JoinFunction;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.windowing.assigners.TumblingProcessingTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.util.Collector;

public class Demo33_CogroupDemo {
    public static void main(String[] args) throws Exception {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        // csv
        // {"id":1,"name":zs,"gender":1}
        DataStreamSource<String> stream1 = env.socketTextStream("doitedu01", 9999);
        SingleOutputStreamOperator<Person> personStream = stream1.map(json -> JSON.parseObject(json, Person.class));


        //{"uid":1,"score":80}
        DataStreamSource<String> stream2 = env.socketTextStream("doitedu01", 9998);
        SingleOutputStreamOperator<Score> scoreStream = stream2.map(s -> JSON.parseObject(s, Score.class));


        DataStream<String> resultStream = personStream.coGroup(scoreStream)
                .where(p -> p.id)
                .equalTo(s -> s.uid)
                .window(TumblingProcessingTimeWindows.of(Time.minutes(1)))
                .apply(new CoGroupFunction<Person, Score, String>() {
                    // 一次给一组（包含左右两流的数据）来调用
                    @Override
                    public void coGroup(Iterable<Person> first, Iterable<Score> second, Collector<String> out) throws Exception {
                        for (Person person : first) {
                            for (Score score : second) {
                                JSONObject jsonObject = new JSONObject();
                                jsonObject.put("student_id", person.id);
                                jsonObject.put("student_name", person.name);
                                jsonObject.put("student_gender", person.gender);
                                jsonObject.put("student_score", score.score);

                                out.collect(jsonObject.toJSONString());

                            }
                        }

                    }
                });


        resultStream.print();

        env.execute();

    }

    @Data
    @NoArgsConstructor
    @AllArgsConstructor
    public static class Person {
        private int id;
        private String name;
        private int gender;
    }


    @Data
    @NoArgsConstructor
    @AllArgsConstructor
    public static class Score {
        private int uid;
        private int score;
    }


}
